package com.github.util;

import com.github.bean.ServerUser;
import com.github.manager.Manager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

public class SocketSendThread {
    // 在控制器或服务中添加测试
    private static final Logger logger = LoggerFactory.getLogger(SocketSendThread.class);

    // 用于存储每个 WebSocketSession 对应的消息队列
    public static final ConcurrentHashMap<String, BlockingQueue<String>> sessionQueues = new ConcurrentHashMap<>();

    // 用于存储每个会话对应的线程引用，便于管理
    private static final ConcurrentHashMap<String, Thread> sessionThreads = new ConcurrentHashMap<>();

    // 用于控制线程退出的标志
    private static final ConcurrentHashMap<String, AtomicBoolean> threadStopFlags = new ConcurrentHashMap<>();

    public static void add(WebSocketSession session) {
        // 为新建立的 WebSocketSession 创建一个消息队列
        BlockingQueue<String> messageQueue = new LinkedBlockingQueue<>();
        sessionQueues.put(session.getId(), messageQueue);

        // 创建线程停止标志
        AtomicBoolean stopFlag = new AtomicBoolean(false);
        threadStopFlags.put(session.getId(), stopFlag);

        // 启动一个可控制的线程来处理该会话的消息队列
        Thread sendThread = new Thread(() -> {
            ServerUser user = null;
            String account = "";
            final String sessionId = session.getId();

            try {
                // 将 while(true) 改为可控制的循环
                while (!stopFlag.get() && !Thread.currentThread().isInterrupted()) {
                    if (user == null) {
                        user = Manager.sessionIDUserMap.getOrDefault(sessionId, null);
                        if (user != null) {
                            account = user.getAccount();
                            synchronized (user) {
                                if (user != null) {
                                    account = user.getAccount();
                                }
                            }
                        }
                    }

                    // 检查会话是否还打开
                    if (!session.isOpen()) {
                        logger.info("会话已关闭，退出发送线程: {}", sessionId);
                        break;
                    }

                    String message = "";
                    try {
                        // 使用 poll 替代 take，设置超时时间避免无限等待
                        message = messageQueue.poll(100, TimeUnit.MILLISECONDS);
                        if (message == null) {
                            // 超时，继续循环检查停止标志
                            continue;
                        }

                        if (!message.contains("\"msg_type\":2")) {
                            logger.info("发送数据-->" + account + ";设备状态:" + session.isOpen() + ";内容-->" + message);
                        }

                        if (session.isOpen()) {
                            // 增加同步块，确保线程安全
                            synchronized (session) {
                                if (session.isOpen()) {
                                    // 发送消息
                                    session.sendMessage(new TextMessage(AESEncrypt.encrypt(message)));
                                } else {
                                    break;
                                }
                            }
                        } else {
                            break;
                        }
                    } catch (InterruptedException e) {
                        // 线程被中断，退出循环
                        Thread.currentThread().interrupt();
                        logger.info("发送线程被中断: {}", sessionId);
                        break;
                    } catch (IOException e) {
                        logger.info("发送数据-->" + account + ";失败,设备状态:" + session.isOpen());
                        // 发生 I/O 异常，检查连接状态
                        if (!session.isOpen()) { // 未打开退出
                            break;
                        }
                        e.printStackTrace();
                        // 重发
                        if (session.isOpen()) {
                            // 增加同步块，确保线程安全
                            synchronized (session) {
                                if (session.isOpen()) {
                                    // 发送消息
                                    try {
                                        session.sendMessage(new TextMessage(AESEncrypt.encrypt(message)));
                                        logger.info("发送数据-->" + account + ";重发成功,设备状态:" + session.isOpen());
                                    } catch (IOException ioException) {
                                        ioException.printStackTrace();
                                        logger.info("发送数据-->" + account + ";重发失败,设备状态:" + session.isOpen());
                                        if (!session.isOpen()) { // 未打开退出
                                            break;
                                        }
                                    }
                                }
                            }
                        }
                    }

                    // 减少CPU占用
                    try {
                        Thread.sleep(1);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        break;
                    }
                }
            } finally {
                logger.info("发送线程退出: {}", sessionId);
                // 清理资源
                cleanupSession(sessionId);
            }
        }, "SocketSendThread-" + session.getId());

        // 设置为守护线程，避免阻止JVM退出
        sendThread.setDaemon(true);
        sessionThreads.put(session.getId(), sendThread);
        sendThread.start();
    }

    public static void remove(WebSocketSession session) {
        String sessionId = session.getId();

        // 设置停止标志
        AtomicBoolean stopFlag = threadStopFlags.get(sessionId);
        if (stopFlag != null) {
            stopFlag.set(true);
        }

        // 中断线程
        Thread thread = sessionThreads.get(sessionId);
        if (thread != null && thread.isAlive()) {
            thread.interrupt();
        }

        // 清理资源
        cleanupSession(sessionId);
    }

    private static void cleanupSession(String sessionId) {
        // 移除消息队列
        sessionQueues.remove(sessionId);
        // 移除线程引用
        sessionThreads.remove(sessionId);
        // 移除停止标志
        threadStopFlags.remove(sessionId);
    }

    public static void sendMessage(WebSocketSession session, String msg) {
        BlockingQueue<String> messageQueue = sessionQueues.get(session.getId());
        if (messageQueue != null) {
            try {
                // 将消息放入队列
                messageQueue.put(msg);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                logger.error("发送消息时被中断", e);
            }
        }
    }
}
